package socketmvc.core.packets;

import socketmvc.core.exception.ResponseException;
import socketmvc.core.packets.basic.IPacket;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.LockSupport;

/**
 *  响应包Future,可用来等待对端的响应报文包
 * @author xueyanjun
 */
public class ResponseFuture extends ControllableFuture<IPacket> {
    // 缓存池,当响应包到达,会从缓存池中获取Future并其设置完成
    protected static final ConcurrentMap<String, ResponseFuture> FUTURE_POOL = new ConcurrentHashMap<>();
    public static ScheduledFuture<?> checkRemoveScheduled;

    static {
        // 自动删除长时间没有收到回复的响应包Future, 避免内存溢出, 目前是超过三分钟没有响应的就认为是废弃的包
        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
        checkRemoveScheduled = scheduledThreadPoolExecutor.scheduleWithFixedDelay(() -> {
            long now = System.currentTimeMillis();
            List<ResponseFuture> removes = new ArrayList<>();
            for (ResponseFuture future : FUTURE_POOL.values()) {
                long timeout = future.getTimeoutAt();
                if (now >= timeout) {
                    removes.add(future);
                }
            }
            if (!removes.isEmpty()) {
                for (ResponseFuture future : removes) {
                    future.cancel(true);
                    FUTURE_POOL.remove(future.getResponseKey());
                }
            }
        }, 1000, 1000 * 60 * 3, TimeUnit.MILLISECONDS);
    }
    // 响应键
    private String responseKey;

    // 当前future的过期移除时间,毫秒值
    private final long timeoutAt;

    public ResponseFuture(String responseKey) {
        super();
        this.responseKey = responseKey;
        this.timeoutAt = this.getCreateAt() + 1000 * 60 * 3;
    }

    public String getResponseKey() {
        return this.responseKey;
    }

    public void setResponseKey(String responseKey) {
        this.responseKey = responseKey;
    }

    public long getTimeoutAt() {
        return timeoutAt;
    }

    @Override
    public IPacket getOrNull(long timeout, TimeUnit unit){
        IPacket pkt = super.getOrNull(timeout, unit);
        if (pkt instanceof ErrorPacket){
            throw new ResponseException(this.pkt.getTextBodyAsStr());
        }
        return pkt;
    }

    @Override
    public IPacket getOrNull() {
        IPacket pkt = super.getOrNull();
        if (pkt instanceof ErrorPacket){
            throw new ResponseException(this.pkt.getTextBodyAsStr());
        }
        return pkt;
    }

    @Override
    public IPacket getOrThrow(long timeout, TimeUnit unit) throws TimeoutException {
        if (this.pkt == null){
            threads.add(Thread.currentThread());
            LockSupport.parkNanos(unit.toNanos(timeout));
            if (this.pkt == null){
                throw new TimeoutException("obtain response timeout, key is :" + this.responseKey);
            }
        }
        if (this.pkt instanceof ErrorPacket){
            throw new ResponseException(this.pkt.getTextBodyAsStr());
        }
        return this.pkt;
    }

    public static void submit(ResponseFuture future){
        FUTURE_POOL.put(future.getResponseKey(),future);
    }

    public static ResponseFuture get(String responseKey){
        return FUTURE_POOL.get(responseKey);
    }

    public static void finish(IPacket packet){
        String responseKey = packet.getResponseKey();
        if (responseKey == null){
            return;
        }
        ResponseFuture responseFuture = ResponseFuture.get(responseKey);
        if (responseFuture == null){
            return;
        }
        responseFuture.set(packet);
    }

}
